/**
 * 
 */
package stc.skymobi.db;

import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.Closure;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import stc.skymobi.ebus.EventBus;


/**
 * @author hp
 *
 */
public class DbVOBatchSplitter implements Closure {

    private static final Logger logger = LoggerFactory.getLogger(DbVOBatchSplitter.class);

    private static class Tuple {
		public	String	fieldName;
		public	int		length = -1;
		
		public Tuple(String value) {
			int idx = value.indexOf(":");
			if ( -1 == idx ) {
				fieldName = value;
			}
			else {
				fieldName = value.substring(0, idx);
				
				try {
					length = Integer.parseInt( value.substring(idx + 1) );
				}
				catch (Exception e) {
					logger.error("Tuple: ", e);
				}
			}
		}
	}
	
    private Map<String, Tuple> properties = new HashMap<String, Tuple>();
    private String  tableName;
    private String  pkIdName;
    private String  pkIdSeqName;
    
    private String  updateSql;
    
    private	ScheduledExecutorService	executorService = null;
    private	List<Object[]>	batchArgs = new LinkedList<Object[]>();
    private	EventBus		eventBus;
    private	String			eventRecordVO;
    private	long			batchMaxDelay = 5000;	// ms, default 5s
    private	int				batchThreshold = 1000;	//	default is 1000
    private	long			lastBatchExecutedTime = 0;
    
    public DbVOBatchSplitter() {
    }
    
    public long getBatchMaxDelay() {
		return batchMaxDelay;
	}

	public void setBatchMaxDelay(long batchMaxDelay) {
		this.batchMaxDelay = batchMaxDelay;
	}

	public EventBus getEventBus() {
		return eventBus;
	}

	public void setEventBus(EventBus eventBus) {
		this.eventBus = eventBus;
	}

	public int getBatchThreshold() {
		return batchThreshold;
	}

	public void setBatchThreshold(int batchThreshold) {
		this.batchThreshold = batchThreshold;
	}

	public String getTable() {
        return	this.tableName;
    }

    /**
     * @param pkIdName the pkIdName to set
     */
    public String getPkId() {
        return	this.pkIdName;
    }

    /**
     * @param pkSeqName the pkSeqName to set
     */
    public String getPkIdSeq() {
        return	this.pkIdSeqName;
    }

    /**
     * @param properties the properties to set
     */
    public Map<String, String> getProperties() {
    	Map<String, String> ret = new HashMap<String, String>();
        for (Map.Entry<String, Tuple> entry : properties.entrySet()) {
        	ret.put(entry.getKey(),  entry.getValue().fieldName);
        }
    	return	ret;
    }
    
    /**
     * @param tableName the tableName to set
     */
    public void setTable(String tableName) {
        this.tableName = tableName;
    }

    /**
     * @param pkIdName the pkIdName to set
     */
    public void setPkId(String pkIdName) {
        this.pkIdName = pkIdName;
    }

    /**
     * @param pkSeqName the pkSeqName to set
     */
    public void setPkIdSeq(String pkIdSeqName) {
        this.pkIdSeqName = pkIdSeqName;
    }

    /**
     * @param properties the properties to set
     */
    public void setProperties(Map<String, String> properties) {
        this.properties.clear();
        
        for (Map.Entry<String, String> entry : properties.entrySet()) {
            if ( null != entry.getValue() ) {
                this.properties.put(entry.getKey(),  new Tuple(entry.getValue()) );
            }
        }
    }

    public void start() {
        StringBuilder   insertSb = new StringBuilder();
        StringBuilder   valuesSb = new StringBuilder();
        
        insertSb.append("insert into ");
        insertSb.append(tableName);
        insertSb.append(" (");
        insertSb.append(pkIdName);
        
        valuesSb.append(" values (");
        valuesSb.append(pkIdSeqName);
        valuesSb.append(".nextval");

        for ( Map.Entry<String, Tuple> entry : properties.entrySet() ) {
            insertSb.append(",");
            insertSb.append(entry.getValue().fieldName);
            
            valuesSb.append(",?");
        }
        
        insertSb.append(") ");
        valuesSb.append(")");
        insertSb.append(valuesSb);
        
        updateSql = insertSb.toString();
        if ( logger.isDebugEnabled() ) {
        	logger.debug("makeupSql:" + updateSql);
        }
        
    	executorService = Executors.newScheduledThreadPool(1, new ThreadFactory(){

			public Thread newThread(Runnable r) {
				return new Thread(r, "DbVOBatchSplitter-thread");
			}});
    	
    	executorService.scheduleWithFixedDelay(new Runnable(){

			public void run() {
				if ( System.currentTimeMillis() - lastBatchExecutedTime > batchMaxDelay ) {
	            	doBatchUpdate();
				}
			}}, 
				0, batchMaxDelay, TimeUnit.MILLISECONDS);
	}
    
    public void stop() {
    	executorService.shutdownNow();
    }
    
    private	void doBatchUpdate() {
    	if ( !batchArgs.isEmpty() ) {
	    	try {
	    		if ( logger.isDebugEnabled() ) {
	    			logger.debug("do batchUpdate" + batchArgs.size() );
	    		}
	    		
	    		final List<Object[]> args = batchArgs;
	    		eventBus.fireEvent(eventRecordVO, updateSql, args);
	    	}
	    	finally {
	    		batchArgs = new LinkedList<Object[]>();
	    		lastBatchExecutedTime = System.currentTimeMillis();
	    	}
    	}
    }
    
    private void doMakeupBatchSQL(Object vo) {
        try {
            Object[] values = new Object[properties.size()];
            int idx = 0;
            for ( Map.Entry<String, Tuple> entry : properties.entrySet() ) {
                Object value = PropertyUtils.getProperty( vo, entry.getKey() );
                int len = entry.getValue().length;
                if ( (value instanceof String) && ( len > 0) 
                		&& ((String)value).length() > len) {
                	value = ((String)value).substring(0, len);
                }
            	values[idx++] = value;
            }
            
            batchArgs.add(values);
            if ( batchArgs.size() >= this.batchThreshold ) {
            	doBatchUpdate();
            }
        } catch (IllegalAccessException e) {
        	logger.error("execute:", e);
        } catch (InvocationTargetException e) {
        	logger.error("execute:", e);
        } catch (NoSuchMethodException e) {
        	logger.error("execute:", e);
        }
    }
    
    /* (non-Javadoc)
     * @see org.apache.commons.collections.Closure#execute(java.lang.Object)
     */
    public void execute(final Object vo) {
    	executorService.submit(new Runnable(){

			public void run() {
				doMakeupBatchSQL(vo);
			}});
    }

	public String getEventRecordVO() {
		return eventRecordVO;
	}

	public void setEventRecordVO(String eventRecordVO) {
		this.eventRecordVO = eventRecordVO;
	}

}
